package com.hefan.monitor.listener;

import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.hefan.common.ons.ONSConstants;
import com.hefan.common.ons.TopicRegistry;
import com.hefan.common.ons.TopicRegistryDev;
import com.hefan.common.ons.TopicRegistryTest;
import com.hefan.monitor.configCenter.MonitorConfigCenter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: ninglijun
 * @date: 15/5/22
 * @time: 下午12:10
 * @description:
 */
@Component
public class ProducerFactory {

    private static final Logger logger = LoggerFactory.getLogger(ProducerFactory.class);

    private static final Map<String, Producer> TOPIC_REL_PRODUCER = new ConcurrentHashMap<String, Producer>();

    @Resource
    private MonitorConfigCenter monitorConfigCenter;

    @PostConstruct
    private void loadProducer() {
        Map<String, String> map = monitorConfigCenter.getPublicConfig();
        String env=map.get("ons.env");
        if (StringUtils.equals(env, "local")) {
            loadTestProducer();
            return;
        }else if(StringUtils.equals(env, "dev")){
            for (String pid : TopicRegistryDev.allProducerIds()) {
                logger.info("loading producer : {}", pid);
                Producer producer = createProducer(pid);
                for (TopicRegistryDev registry : TopicRegistryDev.getTopicsByProducerId(pid)) {
                    TOPIC_REL_PRODUCER.put(registry.getTopic(), producer);
                }
            }
        }else if(StringUtils.equals(env, "test")){
            for (String pid : TopicRegistryTest.allProducerIds()) {
                logger.info("loading producer : {}", pid);
                Producer producer = createProducer(pid);
                for (TopicRegistryTest registry : TopicRegistryTest.getTopicsByProducerId(pid)) {
                    TOPIC_REL_PRODUCER.put(registry.getTopic(), producer);
                }
            }
        }else if(StringUtils.equals(env, "online")){
            for (String pid : TopicRegistry.allProducerIds()) {
                if (StringUtils.equals(pid, TopicRegistry.HEFANTV_TEST2.getProducerId())) {
                    continue;
                }
                logger.info("loading producer : {}", pid);
                Producer producer = createProducer(pid);
                for (TopicRegistry registry : TopicRegistry.getTopicsByProducerId(pid)) {
                    TOPIC_REL_PRODUCER.put(registry.getTopic(), producer);
                }
            }
        }
        for (Map.Entry<String, Producer> entry : TOPIC_REL_PRODUCER.entrySet()) {
            entry.getValue().start();
            logger.info("producer : {} is stared", entry.getKey());
        }
    }

    private void loadTestProducer() {
        String producerId = TopicRegistry.HEFANTV_TEST2.getProducerId();
        Producer producer = createProducer(producerId);
        for (TopicRegistry registry : TopicRegistry.getTopicsByProducerId(producerId)) {
            TOPIC_REL_PRODUCER.put(registry.getTopic(), producer);
        }
    }

    private Producer createProducer(String producerId) {
        Properties prop = new Properties();
        prop.setProperty(PropertyKeyConst.ProducerId, producerId);
        prop.setProperty(PropertyKeyConst.AccessKey, ONSConstants.AccessKey);
        prop.setProperty(PropertyKeyConst.SecretKey, ONSConstants.SecretKey);
        Producer producer = null;
        try {
            producer = com.aliyun.openservices.ons.api.ONSFactory.createProducer(prop);
            logger.info("producer : " + producerId + " loaded");
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("producer : {} load error, error msg : {}", producerId, e.getMessage());
        }

        return producer;
    }

    public Producer getByTopic(String topic) {
        if (TOPIC_REL_PRODUCER.containsKey(topic)) {
            return TOPIC_REL_PRODUCER.get(topic);
        }
        logger.error("topic : {} has no producer, please ensure you topic is correct...", topic);
        return null;
    }
}
